package com.tech.process

import com.tech.common.KuduUtil
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods

//本方法可以合并到向kafka中发送数据时写入
//本方法后续可以补充逻辑，用于对需要发送到kafka的数据做去重校验后再发送到kafka
object NodeProcess {

  def nodeProcess(value: String): Unit = {

    implicit val f: DefaultFormats.type = org.json4s.DefaultFormats
    val map = JsonMethods.parse(value).extract[Map[String, Any]]

    if (map.contains("activity_id")){

      val activity_id = map("activity_id").toString
      val process_id = map("process_id").toString

      if (map.contains("group_id")) {

        val group_id = map("group_id").toString
        KuduUtil.insertPassNodeMsg4Group(group_id, activity_id, process_id)

      } else {

        val customer_id = map("customer_id").toString
        KuduUtil.insertPassNodeMsg(customer_id, activity_id, process_id)

      }
    }

  }

}
